package cn.doitedu.rtmk.demo3;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/22
 * @Desc: 学大数据，上多易教育
 * 规则002：
 * 动态画像条件：  X行为 > 3次；  所以 W行为中的 p1属性的最大值 > p1属性的最小值 * 5
 * 触发事件：     Q行为且属性 p2 > 10
 **/
public class Rule002Calculator implements RuleCalculator {


    ValueState<Integer> cntState;
    ValueState<Double> minState;
    ValueState<Double> maxState;
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext) throws Exception {

        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cntState", Integer.class));
        minState = runtimeContext.getState(new ValueStateDescriptor<Double>("minState", Double.class));
        maxState = runtimeContext.getState(new ValueStateDescriptor<Double>("maxState", Double.class));

        message = new JSONObject();
        message.put("rule_id","rule-002");

    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {


        String eventId = userEvent.getEvent_id();
        long userId = userEvent.getUser_id();

        if("X".equals(eventId)){
            cntState.update(cntState.value() == null? 1 : cntState.value()+1);
        }


        if("W".equals(eventId)){
            String p1ValueStr = userEvent.getProperties().get("p1");
            double p1Value = Double.parseDouble(p1ValueStr);

            if(minState.value() == null || minState.value()> p1Value){
                minState.update(p1Value);
            }

            if(maxState.value() == null || maxState.value() < p1Value){
                maxState.update(p1Value);
            }
        }


        if("Q".equals(eventId) && Double.parseDouble(userEvent.getProperties().get("p2") ) > 10  ){

            if(cntState.value() != null && cntState.value()>3
            && minState!=null & maxState != null
            && maxState.value() > minState.value() * 5){

                message.put("user_id",userId);
                message.put("hit_time",userEvent.getAction_time());

                out.collect(message.toJSONString());
            }

        }


    }
}
